/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.basedt.dms.plugins.datasource.impl.gaussdb;

import cn.hutool.core.util.StrUtil;
import com.basedt.dms.common.constant.Constants;
import com.basedt.dms.plugins.datasource.dto.MaterializedViewDTO;
import com.basedt.dms.plugins.datasource.impl.postgre.PostgreMaterializedViewHandler;
import com.basedt.dms.plugins.datasource.impl.postgre.PostgreObjectTypeMapper;
import com.basedt.dms.plugins.datasource.utils.JdbcUtil;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.List;
import java.util.Objects;

import static com.basedt.dms.plugins.datasource.enums.DbObjectType.MATERIALIZED_VIEW;

public class GaussdbMaterializedViewHandler extends PostgreMaterializedViewHandler {

    @Override
    public List<MaterializedViewDTO> listMViewDetails(String catalog, String schemaPattern, String mViewPattern) throws SQLException {
        String sql = " select " +
                " null as catalog_name," +
                " n.nspname as schema_name," +
                " c.relname as object_name," +
                " 'MATERIALIZED_VIEW'as object_type," +
                " d.description as remark," +
                " v.definition as query_sql," +
                " pg_table_size(concat_ws('.',n.nspname, c.relname)) as data_bytes," +
                " null as create_time," +
                " null as last_ddl_time" +
                " from pg_catalog.pg_namespace n" +
                " join pg_catalog.pg_class c " +
                " on n.oid = c.relnamespace  " +
                " join pg_catalog.gs_matviews v " +
                " on n.nspname = v.schemaname " +
                " and c.relname = v.matviewname " +
                " left join pg_catalog.pg_description d " +
                " on c.oid  = d.objoid " +
                " and d.objsubid = 0" +
                " and d.classoid  = 'pg_class'::regclass" +
                " where c.relkind in ('" + PostgreObjectTypeMapper.mapToOrigin(MATERIALIZED_VIEW) + "')";
        if (StrUtil.isNotEmpty(schemaPattern)) {
            sql += " and n.nspname = '" + schemaPattern + "'";
        }
        if (StrUtil.isNotEmpty(mViewPattern)) {
            sql += " and c.relname = '" + mViewPattern + "'";
        }
        return super.listMViewFromDB(sql);
    }

    @Override
    public String getMViewDdl(String catalog, String schema, String mViewName) throws SQLException {
        if (StrUtil.isEmpty(mViewName)) {
            return "";
        }
        MaterializedViewDTO mv = getMViewDetail(catalog, schema, mViewName);
        if (Objects.nonNull(mv)) {
            StringBuilder ddlBuilder = new StringBuilder();
            if (isIncremental(schema, mViewName)) {
                ddlBuilder.append("CREATE INCREMENTAL MATERIALIZED VIEW ");
            } else {
                ddlBuilder.append("CREATE MATERIALIZED VIEW ");
            }
            ddlBuilder.append(mv.getSchemaName())
                    .append(Constants.SEPARATOR_DOT)
                    .append(mv.getMViewName())
                    .append("\n AS \n");
            if (StrUtil.isNotEmpty(mv.getQuerySql())) {
                ddlBuilder.append(mv.getQuerySql());
            }
            return ddlBuilder.toString();
        } else {
            throw new SQLException(StrUtil.format("materialized view {} does not exist in {}", mViewName, schema));
        }
    }

    private boolean isIncremental(String schema, String mViewName) throws SQLException {
        boolean flag = false;
        String sql = "select " +
                "    n.nspname as schema," +
                "    c.relname as object_name," +
                "    v.ivm as incremental" +
                " from pg_namespace n " +
                " join pg_class c" +
                " on n.oid = c.relnamespace" +
                " join gs_matview v " +
                " on c.oid = v.matviewid" +
                " where n.nspname = ?" +
                " and c.relname = ?";
        Connection conn = dataSource.getConnection();
        PreparedStatement ps = conn.prepareStatement(sql);
        ps.setString(1, schema);
        ps.setString(2, mViewName);
        ResultSet rs = ps.executeQuery();
        while (rs.next()) {
            flag = rs.getBoolean("incremental");
        }
        JdbcUtil.close(conn, ps, rs);
        return flag;
    }
}
